package spring.cloud.tasks.executor.task;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.api.TaskExecuteResultGroup;
import spring.cloud.tasks.executor.common.Service;
import spring.cloud.tasks.executor.listener.config.TaskConfigurationService;
import spring.cloud.tasks.executor.listener.failover.FailoverNode;
import spring.cloud.tasks.executor.listener.report.ReportService;
import spring.cloud.tasks.executor.listener.report.TaskItemExecuteInfo;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.Date;
import java.util.List;

/**
 * 执行作业的服务.
 */
@Slf4j
@Setter
@Getter
public class ExecuteService extends Service {

    private static final String NO_RETURN_VALUE = "No return value.";

    private TaskConfigurationService taskConfigurationService;

    private ReportService reportService;

    public ExecuteService(final TaskExecutor taskExecutor) {
        super(taskExecutor);
    }

    @Override
    public void start() {
        taskConfigurationService = taskExecutor.getTaskConfigurationService();
        reportService = taskExecutor.getReportService();
    }


    /**
     * 注册作业启动信息.
     *
     * @param taskExecuteContext 作业运行时分片上下文
     */
    public void registerTaskItemBegin(final TaskExecuteContext taskExecuteContext) {
        List<Integer> taskItemIndexList = taskExecuteContext.getTaskItemIndexList();
        if (!taskItemIndexList.isEmpty()) {
            reportService.clearInfoMap();
            Date nextFireTimePausePeriodEffected = taskExecutor.getNextFireTimePausePeriodEffected();
            Long nextFireTime = nextFireTimePausePeriodEffected == null ? null : nextFireTimePausePeriodEffected.getTime();
            for (int taskItemIndex : taskItemIndexList) {
                registerTaskItemBegin(taskItemIndex, nextFireTime);
            }
        }
    }

    public void registerTaskItemBegin(int taskItemIndex, Long nextFireTime) {
        if (log.isDebugEnabled()) {
            LogUtils.debug(log, taskId, "registerJobBeginByItem: " + taskItemIndex);
        }
        boolean isEnabledReport = taskConfigurationService.isEnabled();
        if (isEnabledReport) {
            this.getTaskNodeStorage().removeTaskNode(ExecuteNode.getCompletedNode(taskItemIndex));
            this.getTaskNodeStorage().removeTaskNodeIfExisted(ExecuteNode.getFailedNode(taskItemIndex));
            this.getTaskNodeStorage().removeTaskNodeIfExisted(ExecuteNode.getTimeoutNode(taskItemIndex));
            this.getTaskNodeStorage().fillEphemeralTaskNode(ExecuteNode.getRunningNode(taskItemIndex), executorId);
        }

        reportService.initTaskItemExecuteInfo(taskItemIndex, nextFireTime);
    }

    /**
     * 更新当前作业服务器运行时分片的nextFireTime。
     *
     * @param taskItemIndexList 作业运行时分片上下文
     */
    public void updateNextFireTime(final List<Integer> taskItemIndexList) {
        if (!taskItemIndexList.isEmpty()) {
            for (int taskItemIndex : taskItemIndexList) {
                updateNextFireTimeByTaskItemIndex(taskItemIndex);
            }
        }
    }

    private void updateNextFireTimeByTaskItemIndex(int taskItemIndex) {
        if (null == taskExecutor) {
            return;
        }
        Date nextFireTimePausePeriodEffected = taskExecutor.getNextFireTimePausePeriodEffected();
        if (null != nextFireTimePausePeriodEffected) {
            this.getTaskNodeStorage().replaceTaskNode(ExecuteNode.getNextFireTimeNode(taskItemIndex), nextFireTimePausePeriodEffected.getTime());
        }
    }

    /**
     * 判断分片项中是否还有执行中的作业.
     *
     * @param taskItemIndexList 需要判断的分片项列表
     * @return 分片项中是否还有执行中的作业
     */
    public boolean hasRunningTaskItemList(final List<Integer> taskItemIndexList) {
        for (int taskItemIndex : taskItemIndexList) {
            if (this.getTaskNodeStorage().isTaskNodeExisted(ExecuteNode.getRunningNode(taskItemIndex))) {
                return true;
            }
        }
        return false;
    }

    /**
     * 判断是否还有执行中的作业.
     *
     * @return 是否还有执行中的作业
     */
    public boolean hasRunningTaskItemList() {
        return hasRunningTaskItemList(getTaskItemIndexList());
    }

    private List<Integer> getTaskItemIndexList() {
        return Lists.transform(this.getTaskNodeStorage().getTaskNodeChildrenKeys(ExecuteNode.ROOT),
                new Function<String, Integer>() {//
                    @Override
                    public Integer apply(final String input) {
                        return Integer.valueOf(input);
                    }
                });
    }

    public boolean isRunning(final int taskItemIndex) {
        return this.getTaskNodeStorage().isTaskNodeExisted(ExecuteNode.getRunningNode(taskItemIndex));
    }


    //
    public void registerTaskItemCompleted(final TaskExecuteContext taskExecuteContext, int taskItemIndex, Date nextFireTimePausePeriodEffected) throws Exception {
        registerTaskItemCompletedControlInfo(taskExecuteContext, taskItemIndex);
        registerTaskItemCompletedReportInfo(taskExecuteContext, taskItemIndex, nextFireTimePausePeriodEffected);
    }

    /**
     * 注册作业完成信息.
     */
    public void registerTaskItemCompletedControlInfo(final TaskExecuteContext taskExecuteContext, int taskItemIndex) {
        boolean isEnabledReport = taskConfigurationService.isEnabled();
        if (!isEnabledReport) {
            return;
        }

        updateTaskExecuteResult(taskExecuteContext, taskItemIndex);
        // create completed node
        createCompletedNode(taskItemIndex);
        // remove running node
        this.getTaskNodeStorage().removeTaskNode(ExecuteNode.getRunningNode(taskItemIndex));
    }

    private void updateTaskExecuteResult(TaskExecuteContext taskExecuteContext, int taskItemIndex) {
        TaskItemExecuteResult taskExecuteResult = taskExecuteContext.getTaskItemIndexToExecuteResultMap().get(taskItemIndex);
        try {
            if (taskExecuteResult != null) {
                int errorGroup = taskExecuteResult.getErrorGroup();
                if (errorGroup == TaskExecuteResultGroup.TIMEOUT) {
                    this.getTaskNodeStorage().createTaskNodeIfNeeded(ExecuteNode.getTimeoutNode(taskItemIndex));
                } else if (errorGroup != TaskExecuteResultGroup.SUCCESS) {
                    this.getTaskNodeStorage().createTaskNodeIfNeeded(ExecuteNode.getFailedNode(taskItemIndex));
                }
            } else {
                this.getTaskNodeStorage().createTaskNodeIfNeeded(ExecuteNode.getFailedNode(taskItemIndex));
            }
        } catch (RuntimeException e) {
            LogUtils.warn(log, taskId, "update job return fail.", e);
        }
    }

    private void createCompletedNode(int taskItemIndex) {
        try {
            this.getTaskNodeStorage().createOrUpdateTaskNodeWithValue(ExecuteNode.getCompletedNode(taskItemIndex), executorId);
        } catch (RuntimeException e) {
            LogUtils.warn(log, taskId, "update job complete node fail.", e);
        }
    }

    /**
     * 作业完成信息注册，此信息用于页面展现。注意，无论作业是否上报状态(对应/config/enabledReport/节点)，都会注册此信息。
     */
    public void registerTaskItemCompletedReportInfo(final TaskExecuteContext taskExecuteContext, int taskItemIndex, Date nextFireTimePausePeriodEffected) {
        TaskItemExecuteInfo taskItemExecuteInfo = reportService.getTaskItemExecuteInfo(taskItemIndex);
        if (taskItemExecuteInfo == null) { // old data has been flushed to zk.
            taskItemExecuteInfo = new TaskItemExecuteInfo(taskItemIndex);
        }

        TaskItemExecuteResult taskExecuteResult = taskExecuteContext.getTaskItemIndexToExecuteResultMap().get(taskItemIndex);
        if (taskExecuteResult != null) {
            taskItemExecuteInfo.setMessage(taskExecuteResult.getReturnMsg());
            int errorGroup = taskExecuteResult.getErrorGroup();
            //如果作业执行成功且不展示日志，则不展现log
            if (errorGroup == TaskExecuteResultGroup.SUCCESS && !taskConfigurationService.showNormalLog()) {
                taskItemExecuteInfo.setTaskItemLog(null);
            } else {
                taskItemExecuteInfo.setTaskItemLog(taskExecuteContext.getTaskItemLog(taskItemIndex));
            }
        } else {
            taskItemExecuteInfo.setMessage(NO_RETURN_VALUE);
        }
        if (nextFireTimePausePeriodEffected != null) {
            taskItemExecuteInfo.setNextFireTime(nextFireTimePausePeriodEffected.getTime());
        }

        taskItemExecuteInfo.setLastCompleteTime(System.currentTimeMillis());
        reportService.putTaskItemExecuteInfo(taskItemExecuteInfo);
    }

    /**
     * 判断该分片是否已完成.
     *
     * @param taskItemIndex 运行中的分片路径
     * @return 该分片是否已完成
     */
    public boolean isCompleted(final int taskItemIndex) {
        return this.getTaskNodeStorage().isTaskNodeExisted(ExecuteNode.getCompletedNode(taskItemIndex));
    }


    public boolean isFailover(final int taskItemIndex) {
        return this.getTaskNodeStorage().isTaskNodeExisted(FailoverNode.getExecutionFailoverNode(taskItemIndex));
    }

    //
    /**
     * 清除分配分片序列号的运行状态.
     *
     * <p>
     * 用于作业服务器恢复连接注册中心而重新上线的场景, 先清理上次运行时信息.
     * </p>
     *
     * @param items 需要清理的分片项列表
     */
    /**
     * <pre>
     public void clearRunningInfo(final List<Integer> items) {
     for (int each : items) {
     // 已被其他executor接管的正在failover的分片不清理running节点，防止清理节点时触发JobCrashedJobListener导致重新failover了一次
     if (!this.getTaskNodeStorage().isTaskNodeExisted(FailoverNode.getExecutionFailoverNode(each))) {
     this.getTaskNodeStorage().removeTaskNodeIfExisted(ExecuteNode.getRunningNode(each));
     // 清除完成状态timeout等信息

     }
     }
     }
     * </pre>
     */

    /**
     * 删除作业执行时信息.
     */
    public void removeExecutionInfo() {
        this.getTaskNodeStorage().removeTaskNodeIfExisted(ExecuteNode.ROOT);
    }


}
